package com.yanggu.flink.datastream_api.state.keyedstate

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 实现低阶的Regular Joins操作
 * 任何一个流的数据都会触发JOIN操作
 * SELECT * FROM A INNER JOIN B WHERE A.id = B.id;
 *
 * 控制台输出
 * (a,stream-1,1000)=>(a,stream-2,3000)
 * (b,stream-1,2000)=>(b,stream-2,4000)
 */
object ListStateDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val dataStream1 = env
      .fromElements(
        ("a", "stream-1", 1000L),
        ("b", "stream-1", 2000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, Long)] {
          override def extractTimestamp(element: (String, String, Long), recordTimestamp: Long) = element._3
        })
      )


    val dataStream2 = env
      .fromElements(
        ("a", "stream-2", 3000L),
        ("b", "stream-2", 4000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, Long)] {
          override def extractTimestamp(element: (String, String, Long), recordTimestamp: Long) = element._3
        })
      )

    dataStream1
      .keyBy(tuple1 => tuple1._1)
      .connect(dataStream2.keyBy(tuple2 => tuple2._1))
      .process(new TwoStreamJoinProcessFunction)
      .print()

    env.execute()

  }

}

class TwoStreamJoinProcessFunction extends KeyedCoProcessFunction[String, (String, String, Long), (String, String, Long), String] {

  private var stream1ListState: ListState[(String, String, Long)] = _

  private var stream2ListState: ListState[(String, String, Long)] = _

  override def open(parameters: Configuration): Unit = {
    val stream1ListStateDescriptor = new ListStateDescriptor[(String, String, Long)]("stream1ListState", classOf[(String, String, Long)])
    stream1ListState = getRuntimeContext.getListState(stream1ListStateDescriptor)

    val stream2ListStateDescriptor = new ListStateDescriptor[(String, String, Long)]("stream2ListState", classOf[(String, String, Long)])
    stream2ListState = getRuntimeContext.getListState(stream2ListStateDescriptor)

  }

  override def processElement1(value1: (String, String, Long), ctx: KeyedCoProcessFunction[String, (String, String, Long), (String, String, Long), String]#Context, out: Collector[String]): Unit = {
    stream1ListState.add(value1)
    stream2ListState.get().forEach(tempValue2 => out.collect(value1 + "=>" + tempValue2))
  }

  override def processElement2(value2: (String, String, Long), ctx: KeyedCoProcessFunction[String, (String, String, Long), (String, String, Long), String]#Context, out: Collector[String]): Unit = {
    stream2ListState.add(value2)
    stream1ListState.get().forEach(tempValue1 => out.collect(tempValue1 + "=>" + value2))
  }

}
